package com.chenjj.bigdata.zk.client.test;

import com.chenjj.bigdata.zk.leaderelection.Node;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class LeaderElectionTester {

    private static void sleep(long mils) {
        try {
            TimeUnit.MILLISECONDS.sleep(mils);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void test1(){
        final String LEADER_INFO_NODE = "/leader-info";
        int nodeNum = 2;
        AtomicLong idGenerator = new AtomicLong();
        AtomicInteger activeNodeCount = new AtomicInteger();
        while (true) {
            if (activeNodeCount.get() >= nodeNum) {
                sleep(10);
                continue;
            }
            // 线程启动需要一定时间，将线程启动看做开机过程，在开机之前就算一台新的机器加入了
            activeNodeCount.incrementAndGet();
            new Thread(() -> {
                try {
                    Node node = new Node(LEADER_INFO_NODE,"127.0.0.1:2181",1000);
                    while (true) {
                        sleep(5000);
                        // 这里为了试验就让leader有轻微自杀倾向...
                        if (node.getStatus() == Node.Status.LEADER && Math.random() < 0.3) {
                            String logMsg = "----------------------------- " + Thread.currentThread().getName() + " shutdown -----------------------------";
                            System.out.println(logMsg);
                            node.shutdown();
                            break;
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    activeNodeCount.decrementAndGet();
                }
            }, "node-" + idGenerator.getAndIncrement()).start();
        }
    }

    @Test
    public void node1() throws IOException, InterruptedException {
        Node node = new Node("/leader-info","127.0.0.1:2181",1000);
        System.out.println(node.getStatus());
        while(true){
            Thread.sleep(5000);
            System.out.println("node status:"+node.getStatus());
            if (node.getStatus() == Node.Status.LEADER){
                //执行MQ相关逻辑
                System.out.println();
                System.out.println("队列中心正在启动...................");
                System.out.println("队列中心启动成功...................");
            }
        }
    }

    @Test
    public void node2() throws IOException, InterruptedException {
        Node node = new Node("/leader-info","127.0.0.1:2181",1000);
        System.out.println(node.getStatus());
        while(true){
            Thread.sleep(5000);
            System.out.println("node status:"+node.getStatus());
            if (node.getStatus() == Node.Status.LEADER){
                //执行MQ相关逻辑
                System.out.println("队列中心正在启动...................");
                System.out.println("队列中心启动成功...................");
            }
        }
    }
}
